#pragma once

#include <mutex>
#include <condition_variable>
#include <chrono>
#include <queue>
#include <stack>


namespace evm { namespace common {

using namespace std;

/* A multi-producer, multi-consumer queue. */
template <typename ItemType, typename QueueType = queue<ItemType>>
class Queue {
public:
    /* Create a queue object with a given maximum size.
     * If max_size is 0, the queue size is infinite. */
    Queue(size_t max_size = 0) : max_size_(max_size) {}

    /* Clear and reset the queue. */
    void clear() {
        lock_guard<mutex> lg(mutex_);
        queue_ = QueueType();  // new queue means discarding the old one
        unfinished_tasks_ = 0;
    }

    /* Return the approximate size of the queue (not reliable!). */
    size_t size() {
        lock_guard<mutex> lg(mutex_);
        return _size();
    }

    /* Return True if the queue is empty, False otherwise (not reliable!). */
    bool empty() {
        lock_guard<mutex> lg(mutex_);
        return _empty();
    }

    /* Return True if the queue is full, False otherwise (not reliable!). */
    bool full() {
        lock_guard<mutex> lg(mutex_);
        return _full();
    }

    /* Put an item into the queue.
     *
     * If optional args 'block' is true and 'timeout' is 0 (the default),
     * block if necessary until a free slot is available. If 'timeout' is
     * a non-negative number, it blocks at most 'timeout' seconds and return
     * false if no free slot was available within that time.
     * Otherwise ('block' is false), put an item on the queue if a free slot
     * is immediately available, else return false ('timeout'
     * is ignored in that case). */
    bool put(const ItemType& item, bool block = true, size_t timeout = 0) {
        return put(ItemType(item), block, timeout);
    }

    bool put(ItemType&& item, bool block = true, size_t timeout = 0) {
        unique_lock<mutex> ul(mutex_);

        if (block && _full()) {
            if (!timeout)
                not_full_.wait(ul);
            else
                not_full_.wait_for(ul, chrono::seconds(timeout));
        }
        if (_full())
            return false;

        _put(std::move(item));
        unfinished_tasks_ += 1;
        not_empty_.notify_one();
        return true;
    }

    /* Put an item into the queue without blocking.
     *
     * Only enqueue the item if a free slot is immediately available.
     * Otherwise return false. */
    bool put_nowait(const ItemType& item) {
        return put(item, false);
    }

    /* Remove and return an item from the queue.
     *
     * If optional args 'block' is true and 'timeout' is 0 (the default),
     * block if necessary until an item is available. If 'timeout' is
     * a non-negative number, it blocks at most 'timeout' seconds and return
     * default constructed item if no item was available within that time.
     * Otherwise ('block' is false), return an item if one is immediately
     * available, else return default constructed item ('timeout' is ignored in that case). */
    ItemType get(bool block = true, size_t timeout = 0) {
        unique_lock<mutex> ul(mutex_);

        if (block && _empty()) {
            if (!timeout)
                not_empty_.wait(ul);
            else
                not_empty_.wait_for(ul, chrono::seconds(timeout));
        }
        if (_empty())
            return ItemType{};  // default constructed item

        auto item = _get();
        not_full_.notify_one();
        return item;
    }

    /* Remove and return an item from the queue without blocking.
     *
     * Only get an item if one is immediately available. Otherwise return false. */
    ItemType get_nowait() {
        return get(false);
    }

    /* Indicate that a formerly enqueued task is complete.
     *
     * Used by Queue consumer threads.  For each get() used to fetch a task,
     * a subsequent call to task_done() tells the queue that the processing
     * on the task is complete.
     *
     * If a join() is currently blocking, it will resume when all items
     * have been processed (meaning that a task_done() call was received
     * for every item that had been put() into the queue).
     *
     * Raises a ValueError if called more times than there were items
     * placed in the queue. */
    void task_done() {
        unique_lock<mutex> ul(mutex_);

        ptrdiff_t unfinished = unfinished_tasks_ - 1;
        if (unfinished <= 0) {
            if (unfinished < 0)
                throw logic_error("task_done() called too many times");
            all_tasks_done_.notify_all();
        }
        unfinished_tasks_ = static_cast<size_t>(unfinished);
    }

    /* Blocks until all items in the Queue have been gotten and processed.
     *
     * The count of unfinished tasks goes up whenever an item is added to the
     * queue. The count goes down whenever a consumer thread calls task_done()
     * to indicate the item was retrieved and all work on it is complete.
     *
     * When the count of unfinished tasks drops to zero, join() unblocks. */
    void join() {
        unique_lock<mutex> ul(mutex_);
        while (unfinished_tasks_)
            all_tasks_done_.wait(ul);
    }

protected:
    size_t _size() {
        return queue_.size();
    }

    bool _empty() {
        return queue_.empty();
    }

    bool _full() {
        return (max_size_ == 0) ? false : _size() == max_size_;
    }

    void _put(ItemType&& item) {
        queue_.push(std::move(item));
    }

    ItemType _get() {
        auto item = _front();
        queue_.pop();
        return item;
    }

    const ItemType& _front() {
        return queue_.front();
    }

    size_t max_size_;

    QueueType queue_;

    /* mutex must be held whenever the queue is mutating.  All methods
     * that acquire mutex must release it before returning.  mutex
     * is shared between the three conditions, so acquiring and
     * releasing the conditions also acquires and releases mutex. */
    mutex mutex_;

    /* Notify not_empty whenever an item is added to the queue;
     * a thread waiting to get is notified then. */
    condition_variable not_empty_;

    /* Notify not_full whenever an item is removed from the queue;
     * a thread waiting to put is notified then. */
    condition_variable not_full_;

    /* Notify all_tasks_done whenever the number of unfinished tasks
     * drops to zero; thread waiting to join() is notified to resume. */
    condition_variable all_tasks_done_;

    size_t unfinished_tasks_{};
};

/* Variant of Queue that retrieves open entries in priority order (lowest first).
 * Entries are typically tuples of the form:  (priority number, data). */
template <typename ItemType, typename Compare = less<ItemType>>
class PriorityQueue : Queue<ItemType, priority_queue<ItemType, vector<ItemType>, Compare>> {
    const ItemType& _front() override {
        return this->queue_.top();
    }
};

/* Variant of Queue that retrieves most recently added entries first. */
template <typename ItemType>
class LifoQueue : Queue<ItemType, stack<ItemType>> {
    const ItemType& _front() override {
        return this->queue_.top();
    }
};

}}
